package com.xiaozhi.dialogue.stt.providers;

import com.alibaba.nls.client.AccessToken;
import com.alibaba.nls.client.protocol.NlsClient;
import com.alibaba.nls.client.protocol.InputFormatEnum;
import com.alibaba.nls.client.protocol.SampleRateEnum;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriber;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriberListener;
import com.alibaba.nls.client.protocol.asr.SpeechTranscriberResponse;
import com.xiaozhi.dialogue.stt.SttService;
import com.xiaozhi.entity.SysConfig;
import com.xiaozhi.utils.AudioUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Sinks;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

public class AliyunSttNewService implements SttService {
    private static final Logger logger = LoggerFactory.getLogger(AliyunSttNewService.class);
    private static final String PROVIDER_NAME = "aliyun";
    private static final String DEFAULT_MODEL = "paraformer-realtime-v2";

    private final String appKey;
    private final String accessKeyId;
    private final String accessKeySecret;

    private volatile NlsClient client;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "aliyun-token-refresh");
        t.setDaemon(true);
        return t;
    });

    // 最大空闲时间：500毫秒无音频数据则自动停止
    private static final long MAX_IDLE_TIME_MS = 200;
    // 识别超时时间：30秒
    private static final long RECOGNITION_TIMEOUT_MS = 30000;
    // 停止操作超时时间：5秒
    private static final int STOP_TIMEOUT_MS = 5000;

    public AliyunSttNewService(SysConfig config) {
        // 假设SysConfig中有accessKeyId和accessKeySecret字段
        this.appKey = "cayHaQuXAOJyuB4g"; // 需要根据实际字段名调整
        this.accessKeyId = "LTAI5tFSpEuFVeCj8CVqGFCU"; // 需要根据实际字段名调整
        this.accessKeySecret = "Vsvs9UcVfF1kiBL9LDthKg5WwP9aQS"; // 需要根据实际字段名调整

        // 初始化客户端
        initializeClient();

        // 启动 Token 自动刷新（每 30 分钟一次）
        scheduleTokenRefresh();
    }

    /**
     * 初始化 NlsClient 并获取 Token
     */
    private void initializeClient() {
        try {
            AccessToken accessToken = new AccessToken(accessKeyId, accessKeySecret);
            accessToken.apply();
            String token = accessToken.getToken();
            long expireTime = accessToken.getExpireTime();
            logger.info("获取阿里云 ASR Token 成功，有效期至: {}", new java.util.Date(expireTime));

            this.client = new NlsClient(token);
        } catch (Exception e) {
            logger.error("初始化 NlsClient 失败", e);
            this.client = null;
        }
    }

    /**
     * 定时刷新 Token（默认每 30 分钟）
     */
    private void scheduleTokenRefresh() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                logger.debug("正在刷新阿里云 ASR Token...");
                initializeClient();
            } catch (Exception e) {
                logger.warn("自动刷新 Token 失败", e);
            }
        }, 30, 30, TimeUnit.MINUTES);
    }

    @Override
    public String getProviderName() {
        return PROVIDER_NAME;
    }

    @Override
    public boolean supportsStreaming() {
        return true;
    }

    @Override
    public String recognition(byte[] audioData) {
        logger.warn("阿里云单次识别未实现，请使用流式识别");
        return null;
    }

    @Override
    public String streamRecognition(Sinks.Many<byte[]> audioSink) {
        if (client == null) {
            logger.error("NlsClient 未正确初始化，无法进行语音识别");
            return "";
        }

        BlockingQueue<byte[]> audioQueue = new LinkedBlockingQueue<>();
        AtomicBoolean isCompleted = new AtomicBoolean(false);
        AtomicReference<String> finalResult = new AtomicReference<>("");
        CountDownLatch completionLatch = new CountDownLatch(1);
        AtomicBoolean recognitionFinished = new AtomicBoolean(false);
        AtomicBoolean stopSignalSent = new AtomicBoolean(false); // 标记是否已发送停止信号

        // 订阅音频流
        audioSink.asFlux().subscribe(
                chunk -> {
                    try {
                        audioQueue.put(chunk);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        logger.error("添加音频数据到队列时被中断", e);
                    }
                },
                error -> {
                    logger.error("音频流发生错误", error);
                    isCompleted.set(true);
                    completionLatch.countDown();
                },
                () -> {
                    logger.info("音频流结束");
                    isCompleted.set(true);
                    completionLatch.countDown();
                }
        );

        SpeechTranscriber transcriber = null;
        try {
            // 使用数组包装transcriber，使其在匿名内部类中可修改
            final SpeechTranscriber[] transcriberHolder = new SpeechTranscriber[1];
            SpeechTranscriberListener listener = new SpeechTranscriberListener() {
                @Override
                public void onTranscriptionResultChange(SpeechTranscriberResponse response) {
                    logger.info("识别中间结果: {}", response.getTransSentenceText());
                    // 可以在处理中间结果时添加更多上下文信息
                    logger.info("识别中间结果 (taskId: {}): {}", response.getTaskId(), response.getTransSentenceText());
                }

                @Override
                public void onSentenceEnd(SpeechTranscriberResponse response) {
                    String result = response.getTransSentenceText();
                    logger.info("识别完成一句话: {}", result);
                    finalResult.set(result);
                    recognitionFinished.set(true);
                    completionLatch.countDown();
                    // 立即停止识别器，但只在尚未发送停止信号时执行
                    if (transcriberHolder[0] != null && !stopSignalSent.getAndSet(true)) {
                        try {
                            stopTranscriberAsync(transcriberHolder[0]);
                        } catch (Exception e) {
                            logger.warn("停止识别器时出错: {}", e.getMessage());
                        }
                    }
                }

                @Override
                public void onTranscriptionComplete(SpeechTranscriberResponse response) {
                    logger.info("转录完成");
                    // 如果还没有结果，则设置一个空字符串结果
                    if (!recognitionFinished.get()) {
                        recognitionFinished.set(true);
                        completionLatch.countDown();
                    }
                }

                @Override
                public void onTranscriberStart(SpeechTranscriberResponse response) {
                    logger.info("语音识别开始，task_id: {}", response.getTaskId());
                }

                @Override
                public void onSentenceBegin(SpeechTranscriberResponse response) {
                    logger.info("检测到句子开始");
                }

                @Override
                public void onFail(SpeechTranscriberResponse response) {
                    logger.error("语音识别失败: task_id={}, status={}, status_text={}",
                            response.getTaskId(), response.getStatus(), response.getStatusText());
                    recognitionFinished.set(true);
                    completionLatch.countDown();
                    // 立即停止识别器，但只在尚未发送停止信号时执行
                    if (transcriberHolder[0] != null && !stopSignalSent.getAndSet(true)) {
                        try {
                            stopTranscriberAsync(transcriberHolder[0]);
                        } catch (Exception e) {
                            logger.warn("停止识别器时出错: {}", e.getMessage());
                        }
                    }
                }
            };

            transcriber = new SpeechTranscriber(client, listener);
            transcriberHolder[0] = transcriber; // 将transcriber放入holder中
            transcriber.setAppKey(appKey);
            transcriber.setFormat(InputFormatEnum.PCM);
            transcriber.setSampleRate(SampleRateEnum.SAMPLE_RATE_16K);
            transcriber.setEnableIntermediateResult(true);
            transcriber.setEnablePunctuation(true); // 启用标点符号
            transcriber.setEnableITN(true); // 启用反归一化

            // 启动识别
            long startTs = System.currentTimeMillis();
            transcriber.start();
            logger.info("ASR start latency: {} ms", System.currentTimeMillis() - startTs);

            // 发送音频数据
            long lastDataTime = System.currentTimeMillis();
            boolean autoStopped = false;
            int consecutiveEmptyChunks = 0;
            final int MAX_CONSECUTIVE_EMPTY_CHUNKS = 5; // 最大连续空块数
            while (!isCompleted.get() || !audioQueue.isEmpty()) {
                // 如果识别已经完成，立即退出循环
                if (recognitionFinished.get()) {
                    logger.info("识别已完成，退出数据发送循环");
                    break;
                }

                byte[] chunk = audioQueue.poll(100, TimeUnit.MILLISECONDS);
                if (chunk != null) {
                    logger.debug("发送音频数据，长度: {}", chunk.length);
                    transcriber.send(chunk, chunk.length);
                    lastDataTime = System.currentTimeMillis();
                    consecutiveEmptyChunks = 0; // 重置连续空块计数

                    // 精确控制发送节奏（模拟实时流）
                    int sleepTime = calculateSleepTime(chunk.length);
                    if (sleepTime > 0) {
                        Thread.sleep(sleepTime);
                    }
                } else {
                    consecutiveEmptyChunks++;
                    // 如果连续多次获取不到数据，检查是否识别已完成
                    if (consecutiveEmptyChunks >= MAX_CONSECUTIVE_EMPTY_CHUNKS) {
                        if (recognitionFinished.get()) {
                            logger.info("检测到识别完成，退出数据发送循环");
                            break;
                        }
                        consecutiveEmptyChunks = 0; // 重置计数
                    }

                    // 超过最大空闲时间，自动停止识别
                    if (System.currentTimeMillis() - lastDataTime > MAX_IDLE_TIME_MS) {
                        logger.info("音频流空闲超过 {}ms，自动停止识别", MAX_IDLE_TIME_MS);
                        autoStopped = true;
                        break;
                    }
                }
            }

            // 结束识别
            long stopTs = System.currentTimeMillis();
            logger.info("ASR wait for complete");

            // 只有在识别未完成且未发送过停止信号时才发送停止信号
            if (!recognitionFinished.get() && !stopSignalSent.getAndSet(true)) {
                // 如果是自动停止或音频流已完成，发送明确的停止信号
                if (autoStopped || isCompleted.get()) {
                    try {
                        // 发送一个空的数据包作为结束信号
                        transcriber.send(new byte[0], 0);
                    } catch (Exception e) {
                        logger.debug("发送空数据包时出错（可能是正常的）: {}", e.getMessage());
                    }
                }
                stopTranscriberAsync(transcriber);
            } else {
                logger.info("识别已完成或已发送停止信号，跳过停止信号发送");
            }

            // 等待识别完成或超时
            try {
                if (!completionLatch.await(RECOGNITION_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
                    logger.warn("语音识别超时，未在 {}ms 内完成", RECOGNITION_TIMEOUT_MS);
                }
            } catch (InterruptedException e) {
                logger.warn("等待语音识别完成时被中断", e);
                Thread.currentThread().interrupt();
            }

            logger.info("ASR stop latency: {} ms", System.currentTimeMillis() - stopTs);

        } catch (Exception e) {
            logger.error("语音识别过程中发生异常", e);
        } finally {
            // 清理资源
            if (transcriber != null) {
                closeTranscriberAsync(transcriber);
            }
        }

        return finalResult.get();
    }

    /**
     * 异步停止识别器，避免阻塞主线程
     */
    private void stopTranscriberAsync(SpeechTranscriber transcriber) {
        CompletableFuture.runAsync(() -> {
            try {
                transcriber.stop();
            } catch (Exception e) {
                logger.warn("停止识别器时出错: {}", e.getMessage());
            }
        }).orTimeout(STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS).exceptionally(e -> {
            logger.warn("停止识别器超时或出错: {}", e.getMessage());
            return null;
        });
    }

    /**
     * 异步关闭识别器，避免阻塞主线程
     */
    private void closeTranscriberAsync(SpeechTranscriber transcriber) {
        CompletableFuture.runAsync(() -> {
            try {
                transcriber.close();
            } catch (Exception e) {
                logger.warn("关闭识别器时出错: {}", e.getMessage());
            }
        }).orTimeout(STOP_TIMEOUT_MS, TimeUnit.MILLISECONDS).exceptionally(e -> {
            logger.warn("关闭识别器超时: {}", e.getMessage());
            return null;
        });
    }

    /**
     * 根据 PCM 数据大小计算应延迟时间（单位：毫秒）
     * 公式：delay(ms) = (dataSize * 1000) / (sampleRate * bytesPerSample * channels)
     * 假设：16kHz, 16bit(2字节), 单声道
     */
    private int calculateSleepTime(int dataSize) {
        // 16000 Hz × 2 字节/样本 × 1 声道 = 32000 字节/秒
        return (int) (((long) dataSize * 1000) / 32000);
    }

    /**
     * 关闭客户端，释放资源
     */
    public void shutdown() {
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }

        if (client != null) {
            client.shutdown();
            logger.info("NlsClient 已关闭");
        }
    }
}